from datetime import datetime, timedelta
import pymysql
# The DAG object; we'll need this to instantiate a DAG
from airflow import DAG

# Operators; we need this to operate!
# work概念包含了operator和sensor，operators包理解为官方提供，providers理解为第三方提供的工具包（包含了一些operator），可以做一些指定的工作
from airflow.operators.python import PythonOperator
from airflow.operators.email import EmailOperator
from airflow.providers.mysql.operators.mysql import MySqlOperator
from airflow.models import Variable

from airflow.hooks.base import BaseHook

# word = Variable.get("word")
words = Variable.get("words", deserialize_json=True)

def print_word(**context):
    # 获取trigger时候传递进来的配置
    global words
    dag_run_words = context['dag_run'].conf.get('words')
    if dag_run_words:
        words = dag_run_words
    for word in words:
        print(word)
    return words

def save_to_mysql_origin(**context):
    words = context['ti'].xcom_pull(task_ids='print_word')
    conn = BaseHook.get_connection('airflow_test')
    mysqldb = pymysql.connect(host=conn.host, port=conn.port, user=conn.login, passwd=conn.password, db=conn.schema)
    mysqldb.autocommit(True)
    cur = mysqldb.cursor()
    sql = "insert into PrintWordOrigin(`word`) values (%s)"
    effect_row = cur.executemany(sql, words)
    mysqldb.commit()
    # 关闭游标
    cur.close()
    # 关闭连接
    mysqldb.close()

with DAG(
    'print_word',
    tags=['xuwenjie'],
    description='print_word dag',
    default_args={
        'depends_on_past': False,
        # 设置DAG失败或者重试的时候是不是发邮件
        'email': ['1054999089@qq.com'],
        'email_on_failure': True,
        'email_on_retry': True,
        'retries': 1,
        'retry_delay': timedelta(seconds=30),
        # 'queue': 'bash_queue',
        # 'pool': 'backfill',
        # 'priority_weight': 10,
        # 'end_date': datetime(2016, 1, 1),
        # 'wait_for_downstream': False,
        # 'sla': timedelta(hours=2),
        # 'execution_timeout': timedelta(seconds=300),
        # 'on_failure_callback': some_function,
        # 'on_success_callback': some_other_function,
        # 'on_retry_callback': another_function,
        # 'sla_miss_callback': yet_another_function,
        # 'trigger_rule': 'all_success'
    },
    start_date=datetime(2021, 1, 1),
    # schedule_interval=timedelta(days=1),
    schedule_interval='10 22 * * *',
    catchup=False, # 没有被schedule的dag，是否需要回去补跑
    max_active_runs=1, # 最多同时只有一个dag实例在运行，防止数据库死锁
) as dag:

    dag.doc_md = """
    This is dag markdown
    """ 

    print_word = PythonOperator(
        task_id='print_word',
        python_callable=print_word
    )

    save_to_mysql_origin = PythonOperator(
        task_id='save_to_mysql_origin',
        python_callable=save_to_mysql_origin
    )

    summery_word = MySqlOperator(
        task_id= 'summery_word',
        mysql_conn_id = 'airflow_test',
        sql = 'summerywords.sql',
        dag = dag
    )

    email_task = EmailOperator(
        task_id = 'send_email',
        to = '2194766386@qq.com',
        subject = '[DAG] print_word - {{ds}}',
        html_content= '<h3>Success</h3></br>{{conf}}</br>{{execution_date}}',
        dag = dag
    )

    print_word >> save_to_mysql_origin >> summery_word >> email_task